package service

import (
	"strconv"
	"strings"
	"sync"
	"time"
	"vrcm/src/vrcm/app/admin/models"
	"vrcm/src/vrcm/app/admin/models/live_schedule"
	liveDto "vrcm/src/vrcm/app/admin/service/dto/live"
	dto "vrcm/src/vrcm/app/admin/service/dto/live_schedule"
	StreamHubDto "vrcm/src/vrcm/app/admin/service/dto/stream_hub"
	liveService "vrcm/src/vrcm/app/admin/service/live"
	streamHubService "vrcm/src/vrcm/app/admin/service/stream_hub"
	"vrcm/src/vrcm/common/service"
	ext "vrcm/src/vrcm/envconfig"
)

type RecordTaskService struct {
	service.Service
}

func (e *RecordTaskService) DeleteRecordTask(wg *sync.WaitGroup, workTotalNum int, partNum *int, tasks []live_schedule.SourceRecordTask, sourcesMap map[int]live_schedule.SourceBasic, basic live_schedule.ChannelBasic, liveService *liveService.LiveBasic, streamHubService *streamHubService.StreamHubService, liveBasicService SourceBasicService) error {

	var m sync.Mutex
	liveBasicService.Orm = e.Orm
	//// 创建任务通道
	//taskCh := make(chan live_schedule.SourceRecordTask, 3)
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()

	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.SourceRecordTask, wg *sync.WaitGroup) {
			//source := sourcesMap[v.SourceId]
			////删除自定义
			//if source.Kind == 2 && len(source.TaskId) > 0 {
			//	err := liveBasicService.DeleteCustomSource(source)
			//	if err != nil {
			//		e.Log.Errorf("DeleteCustomSource error:%s \r\n", err)
			//	}
			//}
			defer wg.Done()
			//删除协转任务流
			if len(v.RecordTaskId) > 0 {
				err := streamHubService.RemoveStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.RecordTaskId})
				if err != nil {
					e.Log.Errorf("RemoveStreamHub error:%s \r\n", err)
				}
				param := make(map[string]interface{})
				param["end_time"] = now
				err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0", v.RecordTaskId).Updates(&param).Error
				if err != nil {
					e.Log.Errorf("updateScheduleRecord error:%s \r\n", err)
				}
			}
			//删除协转任务流
			if len(v.LiveId) > 0 {
				err := liveService.RemoveLive(&liveDto.StopLiveReq{TenantId: v.TenantId, LiveId: v.LiveId})
				if err != nil {
					e.Log.Errorf("RemoveLive error:%s \r\n", err)
				}
			}
			err := e.Orm.Delete(&v).Error
			if err != nil {
				e.Log.Errorf("SaveWorkflow_Delete error:%s \r\n", err)
			}
			m.Lock()
			*partNum++
			err = e.CalculateProgress(*partNum, workTotalNum, basic)
			if err != nil {
				e.Log.Errorf("Update ChannelTask error:%s \r\n", err)
			}
			e.Log.Trace("当前运行的任务个数：", *partNum, workTotalNum)
			m.Unlock()
		}(tasks[i], wg)
	}
	return nil

}

func (e *RecordTaskService) ProcessSourceRecordTask(wg *sync.WaitGroup, workTotalNum int, partNum *int, tasks []live_schedule.SourceRecordTask, upMap map[int]live_schedule.TrafficOn, basic live_schedule.ChannelBasic, mediaProcess live_schedule.MediaProcess, sourcesMap map[int]live_schedule.SourceBasic) error {
	now := time.Now().Unix()
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	liveService.Log = e.Log
	staticsService := &ChannelStatisticsService{}
	staticsService.Orm = e.Orm
	staticsService.Log = e.Log
	//var m sync.Mutex

	taskCh := make(chan live_schedule.SourceRecordTask, 3)
	// 启动协程执行任务
	go func() {
		for i := 0; i < len(tasks); i++ {
			taskCh <- tasks[i] // 将任务放入通道
		}
		close(taskCh) // 关闭通道，表示任务已经全部放入通道
	}()

	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.SourceRecordTask, wg *sync.WaitGroup) {
			defer wg.Done()
			//创建协转任务
			//获取信源基本信息
			sourceBasic, ok := sourcesMap[v.SourceId]
			if !ok {
				return
			}
			if v.RecordState == 1 {
				return
			}

			if len(v.RecordTaskId) > 0 {
				return
			}
			//全球直播获取对应的拉流地址
			recordTaskId := ""
			//如果没有录制任务需要重新创建录制任务
			streamReq := GenerateInsertLiveParam(&dto.SourceBasicInsertReq{TenantId: v.TenantId, SourceName: sourceBasic.SourceName})
			streamReq.AutoRecord = "1"
			//1.获取全球直播推流地址
			streamReq.RecordFileCallbackUrl = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/source-basic/record-file-call-back"
			streamReq.Tag = strconv.Itoa(sourceBasic.Id)
			res, err := liveService.CreateLiveStream(streamReq)
			if err != nil {
				e.Log.Errorf("CreateLiveStream error:%s \r\n", err)
			}
			if res != nil {
				v.SourceId = sourceBasic.Id
				v.TenantId = sourceBasic.TenantId
				v.CreateTime = now
				v.UpdateTime = now
				v.LiveId = res.LiveId
				v.RecordType = 2
				streamType := "rtmp"
				pushes := strings.Split(res.PushUrl, models.Delimiter)
				if len(pushes) > 0 {
					for _, p := range pushes {
						if strings.Contains(p, streamType) {
							v.PushUrl = p
						}
					}
				}
			}

			//3.创建协转任务
			sources := make([]StreamHubDto.Source, 0, 1)
			url := sourceBasic.OutUdpUrl
			source := StreamHubDto.Source{
				Url: url,
			}
			sources = append(sources, source)

			outs := make([]StreamHubDto.Target, 0, 1)
			out := StreamHubDto.Target{
				Format: "rtmp",
				Url:    v.PushUrl,
				Type:   "push",
			}
			outs = append(outs, out)
			req := GenerateStreamHubReq(sources, outs, sourceBasic.TenantId, sourceBasic.SourceName, "udp", "push", 0)
			//最后调用hub接口
			hub, err := streamHubService.CreateStreamHub(req)
			if err != nil {
				e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
				return
			}

			if hub != nil {
				v.RecordTaskId = hub.TaskId
				recordTaskId = hub.TaskId
			}
			//创建录制任务
			if sourceBasic.SourceState == 3 {
				v.RecordState = 3
			} else {
				v.RecordState = 1
			}

			err = e.Orm.Model(live_schedule.SourceRecordTask{}).Create(&v).Error
			if err != nil {
				e.Log.Errorf("CreateSourceRecordTask error:%s \r\n", err)
				return
			}
			e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Find(&basic)
			if basic.ChannelState != 2 {
				//开启录制
				err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: sourceBasic.TenantId, Id: recordTaskId})
				if err != nil {
					e.Log.Errorf("StartStreamHub error:%s \r\n", err)
					return
				}

				//创建调度统计记录
				err = staticsService.CreateScheduleRecord(sourceBasic.TenantId, v.ChannelId, recordTaskId, 3, now)
				if err != nil {
					e.Log.Errorf("CreateScheduleRecord error:%s \r\n", err)
				}
			}

			//m.Lock()
			//*partNum++
			//err = e.CalculateProgress(*partNum, workTotalNum, basic)
			//if err != nil {
			//	e.Log.Trace("CalculateProgress error：", workTotalNum)
			//}
			//e.Log.Trace("当前运行的任务个数：", partNum, workTotalNum)
			//m.Unlock()
		}(tasks[i], wg)
	}

	return nil
}

func (e *RecordTaskService) StartRecordTasks(wg *sync.WaitGroup, tasks []live_schedule.SourceRecordTask, workNum int, partNum *int, param map[string]interface{}, sourceMap map[int]live_schedule.SourceBasic) error {
	// 创建任务通道
	//taskCh := make(chan live_schedule.SourceRecordTask, 3)
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	sourceBasicService := &SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	channelStatisticsService := &ChannelStatisticsService{}
	channelStatisticsService.Log = e.Log
	channelStatisticsService.Orm = e.Orm
	//// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()
	////var mu sync.Mutex
	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.SourceRecordTask, wg *sync.WaitGroup) {
			//up := sourceMap[v.SourceId]
			//if up.Kind == 2 {
			//	err := sourceBasicService.StartSource(&dto.StartSourceReq{TenantId: v.TenantId, TaskId: up.TaskId, Id: v.SourceId})
			//	if err != nil {
			//		e.Log.Errorf("StartSource error:%s \r\n", err)
			//	}
			//}
			defer wg.Done()
			if len(v.RecordTaskId) > 0 {
				//开启协助任务
				err := streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.RecordTaskId})
				if err != nil {
					e.Log.Errorf("StartStreamHub error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
				//记录任务启动记录用于统计
				err = channelStatisticsService.CreateScheduleRecord(v.TenantId, v.ChannelId, v.RecordTaskId, 3, now)
				if err != nil {
					e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
				}
			}
			//mu.Lock()
			//*partNum++
			//result := float64(*partNum) / float64(workNum)
			//formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
			//e.Log.Trace("当前进度为", formattedResult)
			//if workNum == *partNum {
			//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Updates(param).Error
			//	if err != nil {
			//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
			//	}
			//} else {
			//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Update("progress", formattedResult).Error
			//	if err != nil {
			//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
			//	}
			//}
			//e.Log.Trace("当前运行的任务个数：", *partNum, workNum)
			//mu.Unlock()
		}(tasks[i], wg)
	}

	return nil
}

func (e *RecordTaskService) StopRecordTasks(wg *sync.WaitGroup, tasks []live_schedule.SourceRecordTask, workNum int, partNum *int, param map[string]interface{}, sourceMap map[int]live_schedule.SourceBasic) error {
	// 创建任务通道
	//taskCh := make(chan live_schedule.SourceRecordTask, 3)
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	sourceBasicService := &SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	//// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()

	//var mu sync.Mutex
	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.SourceRecordTask, wg *sync.WaitGroup) {
			//up := sourceMap[v.SourceId]
			//if up.Kind == 2 {
			//	err := sourceBasicService.StopSource(&dto.StopSourceReq{TenantId: v.TenantId, TaskId: up.TaskId})
			//	if err != nil {
			//		e.Log.Errorf("StopSource error:%s \r\n", err)
			//	}
			//}
			defer wg.Done()
			if len(v.RecordTaskId) > 0 {
				//停止录制
				err := streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.RecordTaskId})
				if err != nil {
					e.Log.Errorf("StopStreamHub error:%s \r\n", err)
				}

				//更新任务记录结束时间
				param2 := make(map[string]interface{})
				param2["end_time"] = now
				err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and type =3", v.RecordTaskId).Updates(&param2).Error
				if err != nil {
					e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
				}
			}
			//更新通道任务状态为离线
			param1 := make(map[string]interface{})
			param1["update_time"] = now
			param1["record_state"] = 2
			err := e.Orm.Model(live_schedule.SourceRecordTask{}).Where("tenant_id=? and channel_id=?", v.TenantId, v.ChannelId).Updates(&param1).Error
			if err != nil {
				e.Log.Errorf("ProcessWorkflow Updates SourceRecordTask error:%s \r\n", err)
			}

			//mu.Lock()
			//*partNum++
			//result := float64(*partNum) / float64(workNum)
			//formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
			//e.Log.Trace("当前进度为", formattedResult)
			//if *partNum == workNum {
			//	err := e.Orm.Model(live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Updates(param).Error
			//	if err != nil {
			//		e.Log.Errorf("Update ChannelTask error:%s \r\n", err)
			//	}
			//} else {
			//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Update("progress", formattedResult).Error
			//	if err != nil {
			//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
			//	}
			//}
			//e.Log.Trace("当前运行的任务个数：", *partNum, workNum)
			//mu.Unlock()
		}(tasks[i], wg)
	}

	return nil

}

func GenerateStreamHubReq(sources []StreamHubDto.Source, outs []StreamHubDto.Target, tenantId string, name string, inputFormat string, inputType string, delayTime int) *StreamHubDto.StreamHubReq {
	req := &StreamHubDto.StreamHubReq{}
	req.Name = name
	req.Type = "live"
	req.TenantId = tenantId
	req.Description = ""
	req.Input = StreamHubDto.Input{
		Format: inputFormat,
		Active: 0,
		Policy: StreamHubDto.Policy{
			Mode: "manual",
		},
		Type:    inputType,
		Sources: sources,
	}

	delay := StreamHubDto.Delay{
		Enable:   "no",
		Duration: 10,
	}
	if delayTime > 0 {
		delay.Enable = "yes"
		delay.Duration = delayTime
	}
	req.Output = StreamHubDto.Output{
		Targets: outs,
		Delay:   delay,
		Acl:     StreamHubDto.Acl{},
	}
	return req
}

func (e *RecordTaskService) CalculateProgress(partNum int, workTotalNum int, basic live_schedule.ChannelBasic) error {
	result := float64(partNum) / float64(workTotalNum)
	formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
	e.Log.Trace("当前进度为", formattedResult)
	param := make(map[string]interface{})
	param["progress"] = formattedResult
	if workTotalNum <= partNum {
		param["channel_state"] = basic.ChannelState
	}
	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Updates(&param).Error
	if err != nil {
		e.Log.Trace("update ChannelBasic error：", err)
	}
	return err
}
